package storm;


import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.PrintStream;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import twitter4j.Status;
import utilities.Twokenizer;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class TokenizerBolt extends BaseBasicBolt {

	private static final long serialVersionUID = 4493364795064805398L;
	private List<String> list = new LinkedList<String>();

	
	@Override
	public void execute(Tuple input, BasicOutputCollector collector) {
		if(((Status)input.getValueByField("tweet")).getIsoLanguageCode().equals("en")){
			Twokenizer tokenizer = new Twokenizer();
			List<String> tokens = tokenizer.twokenize(((Status)input.getValueByField("tweet")).getText());
			String tmp = "";
			String tweetTok = "";
			Iterator<String> iter = tokens.iterator();
			while(iter.hasNext()){
				tmp = (String)iter.next();
				if(tmp.startsWith("#")||tmp.startsWith("@")||tmp.contains("http")){
					tweetTok = tweetTok + " " + tmp;
				}
				else{
					tmp = tmp.replaceAll("[^a-zA-Z0-9\\s]", "");
					tweetTok = tweetTok + " " + tmp;
				}
			}
			collector.emit(new Values(tweetTok.toLowerCase()));
			list.add(tmp);
		}
	}
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("tweetTok"));
	}

	public void cleanup() {
		createFile();
				System.out.println("CREATED DUMP FILE");
		
	}
	private void createFile(){
		PrintStream ps=null;
		try {
			ps = new PrintStream(new FileOutputStream("output.txt"));
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		for(String s : list){
			ps.println(s);
		}
		ps.close();
	}
}
